'''
@Company: TWL
@Author: xue jian
@Email: xuejian@kanzhun.com
@Date: 2020-07-20 15:39:30
'''
from transfer import *
from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka import TopicPartition
if __name__ == "__main__":
    transfer = Transfer()
    bootstrap_servers=['172.21.211.126:9092','172.21.211.82:9092','172.21.211.41:9092','172.21.211.39:9092','172.21.211.214:9092']
    # producer_bootstrap_servers=['172.21.32.178:9092', '172.21.32.154:9092', '172.21.32.41:9092', '172.21.32.50:9092', '172.21.32.98:9092']
    topic = "boss.arc.recommender.geekrec_sample_flow"
    # topic = 'boss.arc.recommender.bossrec_sample_flow'
    consumer = KafkaConsumer(topic, auto_offset_reset='latest', group_id='transfer', bootstrap_servers=bootstrap_servers)
    producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
    produce_topic = 'boss.arc_six.rcd.geekrec_fid_flow_new'
    # produce_topic = 'boss.arc_six.rcd.rank_chat_fid_flow'
    i = 0
    for msg in consumer:
        # print(msg.value)
        msg = json.loads(msg.value.decode())
        p_msg = transfer.transfer_c_data(msg)
        if p_msg == None:
            # print('None')
            continue
        try:
            # print(p_msg)
            producer.send(produce_topic, p_msg.encode(encoding='utf-8'))
        except:
            print('transfer error!!!')
            continue
        i += 1
        if i > 6000000:
            print('consumer 2 million data')
            i = 0
